#!/usr/bin/python3
"""vp-demo: the engine of Vision Test Drive, an Apple-Store-style demo remote for Jason's Apple Vision Pro (M5).
(The engine keeps its internal name vp-demo: other lanes and saved state call it by that name.)

What it does: serves one touch page on this Mac's home network. Tapping a tile on an iPhone,
iPad or the Mac opens that app on the headset through Xcode's developer connection
(`xcrun devicectl device process launch`). Nothing is installed on the headset and no
setting changes; it only launches apps that are already there.

Jason 2026-10-17 8:30am: "I want and need the abililty to launch apps from an iphone, ipad,
and mac so I can contorl demo experiences similar to what they do in the apple store demos
themselves."

Usage:
  vp-demo serve [--port 8787]       run the remote (the Vision Test Drive menu bar app runs this for you)
  vp-demo launch <name or bundle>   open one app on the headset from the command line
  vp-demo list                      the tiles and every installed app it knows
  vp-demo refresh                   re-read the headset's app list into the cache
  vp-demo reset                     close every app a guest opened (and Photos)
  vp-demo code                      the six-digit pairing code for a new phone, iPad or browser
  vp-demo devices                   paired devices;  vp-demo unpair-all  forgets them all
  vp-demo headsets                  the headsets paired with this Mac;  vp-demo use <name>  picks one
  vp-demo agent                     (re)install the fallback LaunchAgent (only without the app)
  vp-demo --selftest                runs the real server on a spare port with a stand-in headset

Pairing: each device pairs once with a six-digit code (valid ten minutes, single use, five wrong
tries per address per ten minutes) and then holds its own token in an HttpOnly cookie. The master
key in ~/.config/vp-demo/key never leaves the Mac; the menu bar app sends it as an X-Key header.
"""
import contextlib, fcntl, html, json, os, re, secrets, subprocess, sys, threading, time, urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

CFG = os.path.expanduser("~/.config/vp-demo")
HERE = os.path.dirname(os.path.abspath(__file__))
APPS_CACHE = os.path.join(CFG, "apps.json")
LOG = os.path.join(CFG, "launches.log")
TIMEOUT = int(os.environ.get("VP_DEMO_TIMEOUT", "45"))

# The demo set, in the order a demo runs. Names match the headset's own app names.
# The demo set anyone gets on first run: Apple's own showpieces and public App Store apps, in
# the order a demo runs. The owner's own set lives in ~/.config/vp-demo/tiles.json (load_demo),
# so a download never carries anybody else's apps or family names.
DEFAULT_DEMO = [
    ("Start here", [("Photos", "com.apple.mobileslideshow"), ("Spatial Gallery", "com.apple.SpatialGallery"),
                    ("TV (Apple Immersive)", "com.apple.tv"), ("Encounter Dinosaurs", "com.apple.encounterdinosaurs"),
                    ("Mindfulness", "com.apple.Mind")]),
    ("Apple Immersive titles", [("Elevated", "title:elevated"), ("Boundless", "title:boundless"),
                                ("Real Madrid", "title:realmadrid"), ("BBC Proms", "title:proms")]),
    ("Travel and places", [("FLY", "com.VirZOOM.FlyVP"), ("Epic Earth", "com.mff.ImmersiveFilms"),
                           ("Explore POV", "com.hustler.james.ExplorePOV"), ("Zillow Immerse", "com.zillow.ListingVision"),
                           ("Sky Guide", "com.fifthstarlabs.skyguide")]),
    ("Showpieces", [("What If...?", "com.disney.whatif"), ("Puzzling Places", "com.Realities.io-Inc.Puzzling-Places-AVP"),
                    ("JigSpace", "com.jigspace.jigspace.vision"), ("Disney+", "com.disney.disneyplus"),
                    ("NBA", "com.nbaimd.gametime.universal")]),
]
TILES_FILE = os.path.join(CFG, "tiles.json")


def load_demo(path=None):
    """The owner's tile list from tiles.json ([[section, [[name, id], ...]], ...]); DEFAULT_DEMO when
    the file is missing or unreadable. A tile whose id fails valid_bundle is dropped, never launched."""
    try:
        raw = json.load(open(path or TILES_FILE))
        out = []
        for section, tiles in raw:
            keep = [(str(n), str(b)) for n, b in tiles if valid_bundle(str(b))]
            if keep:
                out.append((str(section), keep))
        return out or DEFAULT_DEMO
    except Exception:
        return DEFAULT_DEMO


def device():
    """Which headset: VP_DEMO_DEVICE, else the id saved in ~/.config/vp-demo/device, else the first
    PHYSICAL visionOS device devicectl knows (simulators are skipped), saved for next time."""
    env = os.environ.get("VP_DEMO_DEVICE")
    if env:
        return env
    f = os.path.join(CFG, "device")
    try:
        d = open(f).read().strip()
        if d:
            return d
    except OSError:
        pass
    hs = headsets()                            # two headsets: wait for `vp-demo use`, never guess
    if len(hs) == 1:
        save_device(hs[0][0], hs[0][1])        # the name too, so re-discovery can tell it apart later
        return hs[0][0]
    return "no-vision-pro-paired"


def headsets(listing=None):
    """(identifier, name) for every physical headset devicectl knows."""
    if listing is None:
        tmp = os.path.join(CFG, "devices-scan.json")
        os.makedirs(CFG, exist_ok=True)
        try:
            subprocess.run(["xcrun", "devicectl", "list", "devices", "--json-output", tmp], capture_output=True, timeout=30)
            listing = json.load(open(tmp))
        except Exception:
            return []
    return [(x.get("identifier"), x.get("deviceProperties", {}).get("name") or x.get("identifier"))
            for x in listing.get("result", {}).get("devices", [])
            if x.get("hardwareProperties", {}).get("platform") == "visionOS"
            and x.get("hardwareProperties", {}).get("reality") == "physical" and x.get("identifier")]


def discover_device(listing=None, single=False):
    """The first physical visionOS device in a devicectl device listing (read live when none given).
    single=True answers only when exactly one physical headset is listed."""
    if listing is None:
        tmp = os.path.join(CFG, "devices-scan.json")
        os.makedirs(CFG, exist_ok=True)
        try:
            subprocess.run(["xcrun", "devicectl", "list", "devices", "--json-output", tmp], capture_output=True, timeout=30)
            listing = json.load(open(tmp))
        except Exception:
            return None
    real = [x.get("identifier") for x in listing.get("result", {}).get("devices", [])
            if x.get("hardwareProperties", {}).get("platform") == "visionOS"
            and x.get("hardwareProperties", {}).get("reality") == "physical" and x.get("identifier")]
    if single:
        return real[0] if len(real) == 1 else None
    return real[0] if real else None


# Titles open inside an app at one show. The page sends only the "title:" key, never a URL, so the
# server launches nothing it did not list here.
TITLES = {
    "title:elevated": ("com.apple.tv", "https://tv.apple.com/us/show/elevated/umc.cmc.5klf6wzriakqz7lcarzyp5bp1"),
    "title:boundless": ("com.apple.tv", "https://tv.apple.com/us/show/boundless/umc.cmc.2y2ypiwhg8dp8027p47jfhlf1"),
    "title:realmadrid": ("com.apple.tv", "https://tv.apple.com/us/movie/real-madrid-the-weight-of-greatness/umc.cmc.59n2yrotl8j6izhn13cj15kz9"),
    "title:proms": ("com.apple.tv", "https://tv.apple.com/us/movie/debut-at-the-bbc-proms/umc.cmc.12u6qljhkbg7up6p3oehb0gbs"),
}


def load_apps():
    try:
        d = json.load(open(APPS_CACHE))
    except Exception:
        return {}
    return {a["bundleIdentifier"]: a.get("name", a["bundleIdentifier"]) for a in d.get("result", {}).get("apps", [])
            if a.get("bundleIdentifier")}


def rediscover():
    """After Xcode re-pairs the headset its id changes: when exactly one physical headset is listed
    and it is not the saved one, save the new id. Two or more headsets leave the saved choice alone."""
    if os.environ.get("VP_DEMO_DEVICE"):
        return
    hs = headsets()
    if len(hs) != 1:
        return
    found, name = hs[0]
    try:
        cur = open(os.path.join(CFG, "device")).read().strip()
    except OSError:
        cur = ""
    try:
        cur_name = open(os.path.join(CFG, "device-name")).read().strip()
    except OSError:
        cur_name = ""
    # Follow the same headset (same name) to its new id after a re-pair; never swap to a different
    # headset because the chosen one is switched off right now. A saved id with no name (older
    # installs) gets its name the next time that headset is listed, and is never swapped meanwhile.
    if found == cur and not cur_name:
        save_device(found, name)
    elif found != cur and (not cur or cur_name == name):
        save_device(found, name)


def save_device(ident, name=""):
    """Write the chosen headset atomically (id, and its name when known)."""
    os.makedirs(CFG, exist_ok=True)
    for fname, val in (("device", ident), ("device-name", name)):
        if fname == "device-name" and not val:
            continue
        tmp = os.path.join(CFG, f"{fname}.{os.getpid()}.tmp")
        open(tmp, "w").write(val + "\n")
        os.replace(tmp, os.path.join(CFG, fname))


_refresh_lock = threading.Lock()


def refresh():
    """Re-read the headset's app list into the cache. Never raises; one re-read at a time."""
    if not _refresh_lock.acquire(blocking=False):
        return False, ["a re-read is already running"]
    try:
        os.makedirs(CFG, exist_ok=True)
        rediscover()
        tmp = APPS_CACHE + ".tmp"
        try:
            r = subprocess.run(["xcrun", "devicectl", "device", "info", "apps", "--include-all-apps", "--device", device(),
                                "--timeout", str(TIMEOUT), "--json-output", tmp], capture_output=True, text=True, timeout=TIMEOUT + 30)
        except (subprocess.TimeoutExpired, OSError) as e:
            return False, [f"no answer from the headset ({type(e).__name__})"]
        try:
            ok = r.returncode == 0 and len(json.load(open(tmp)).get("result", {}).get("apps", [])) > 0
        except Exception:
            ok = False
        if not ok:
            try:
                os.remove(tmp)
            except OSError:
                pass
            return False, (r.stderr or r.stdout).strip().splitlines()[-1:] or ["no output"]
        os.replace(tmp, APPS_CACHE)
        return True, [f"{len(load_apps())} apps"]
    finally:
        _refresh_lock.release()


def refresh_soon():
    """Start a background re-read (Reset with a missing list, the Mac's Refresh item)."""
    threading.Thread(target=refresh, daemon=True).start()


def resolve(target, apps):
    if target in apps:
        return target
    t = target.lower()
    for b, n in apps.items():
        if n.lower() == t:
            return b
    for _, tiles in load_demo():
        for n, b in tiles:
            if n.lower() == t or n.lower().startswith(t):
                return b
    return None


def valid_bundle(b):
    """A bundle id is letters, digits, dots, hyphens and underscores only: nothing a shell could read."""
    if b in TITLES:
        return True
    return bool(b) and len(b) < 200 and all(c.isalnum() or c in "._-" for c in b) and b[0].isalnum()


def allowed(b):
    """Launch policy, one for the page, the Mac app and the CLI: a listed title, a demo tile, or an
    app the headset reports as installed. Anything else is refused before devicectl sees it."""
    if not valid_bundle(b):
        return False
    if b in TITLES or any(b == t for _, tiles in load_demo() for _, t in tiles):
        return True
    return b in load_apps()


_lock = threading.Lock()
_last = {"id": None}                          # the tile last opened from any remote, until the next Reset
ICONS = os.path.join(CFG, "icons")


@contextlib.contextmanager
def device_turn():
    """One launch or reset at a time on this Mac: a thread lock for the engine's own requests plus a
    file lock, so a standalone `vp-demo launch` and the engine never talk to the headset at once.
    Yields False (and touches nothing) when someone else has the turn."""
    if not _lock.acquire(blocking=False):
        yield False
        return
    fd = None
    try:
        os.makedirs(CFG, exist_ok=True)
        fd = os.open(os.path.join(CFG, "device.lock"), os.O_RDWR | os.O_CREAT, 0o600)
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            yield False
            return
        yield True
    finally:
        if fd is not None:
            os.close(fd)                         # closing releases the file lock
        _lock.release()


_state_cache = {"at": 0.0, "value": None}
_state_lock = threading.Lock()


def cached_lock_state(max_age=8):
    """The status pill and the menu bar poll this. Every open page and the Mac app share one reading
    at most `max_age` seconds old, so ten open remotes cost one devicectl call, not ten (the processor
    runs out first on these Macs: hardware-classes.md). A launch always takes a fresh reading."""
    with _state_lock:
        if _state_cache["value"] is None or time.time() - _state_cache["at"] > max_age:
            _state_cache["value"] = lock_state()
            _state_cache["at"] = time.time()
        return _state_cache["value"]


_xcode = {"ok": None, "at": 0.0}


def has_devicectl():
    """Xcode's device tools are what open apps on the headset. Checked once a minute at most."""
    if _xcode["ok"] is None or time.time() - _xcode["at"] > 60:
        try:
            _xcode["ok"] = subprocess.run(["xcrun", "--find", "devicectl"], capture_output=True, timeout=10).returncode == 0
        except Exception:
            _xcode["ok"] = False
        _xcode["at"] = time.time()
    return _xcode["ok"]


def lock_state():
    """ready / locked / unreachable / noxcode, in about a second. A locked headset (off his head)
    cannot open apps: devicectl then hangs past its own timeout, so check this before every launch.
    A reading without the passcodeRequired field is unknown, never ready."""
    if not has_devicectl():
        return "noxcode"
    tmp = os.path.join(CFG, f"lock-{threading.get_ident()}.json")
    os.makedirs(CFG, exist_ok=True)
    try:
        r = subprocess.run(["xcrun", "devicectl", "device", "info", "lockState", "--device", device(), "--timeout", "12",
                            "--json-output", tmp], capture_output=True, text=True, timeout=25)
        res = (json.load(open(tmp)).get("result") or {}) if r.returncode == 0 else {}
    except Exception:
        res = {}
    finally:
        try:
            os.remove(tmp)
        except OSError:
            pass
    return parse_lock(res)


def parse_lock(res):
    if not isinstance(res, dict) or not isinstance(res.get("passcodeRequired"), bool):
        return "unreachable"
    return "locked" if res["passcodeRequired"] else "ready"


STATE_TEXT = {"ready": "Headset ready", "locked": "Headset locked: put it on to unlock",
              "unreachable": "The Mac cannot reach the headset (is it on and on this Wi-Fi?)",
              "noxcode": "Install Xcode on this Mac, then pair the headset once in Xcode's Devices and Simulators window"}


def fetch_icon(bundle):
    """Cache the app's real icon from the headset (160 px). Returns the path or None."""
    os.makedirs(ICONS, exist_ok=True)
    out = os.path.join(ICONS, bundle + ".png")
    if os.path.exists(out) and os.path.getsize(out) > 0:
        if os.path.getsize(out) > 60000:
            subprocess.run(["sips", "-Z", "160", out], capture_output=True)
        return out
    tmp = out + ".tmp.png"
    try:
        subprocess.run(["xcrun", "devicectl", "device", "info", "appIcon", "--device", device(), "--app-bundle-id", bundle,
                        "--width", "160", "--height", "160", "--scale", "1", "--destination", tmp, "--timeout", "20"],
                       capture_output=True, text=True, timeout=35)
    except subprocess.TimeoutExpired:
        pass
    if os.path.exists(tmp) and os.path.getsize(tmp) > 0:
        # devicectl returns 1024 px whatever --width says; 160 px keeps a 150-icon page light on a phone.
        # Resized aside and swapped in whole, so a page never fetches a half-written icon.
        subprocess.run(["sips", "-Z", "160", tmp], capture_output=True)
        os.replace(tmp, out)
        return out
    try:
        os.remove(tmp)
    except OSError:
        pass
    return None


def warm_icons():
    """Background: fetch every tile's icon once, demo tiles first."""
    seen = []
    for _, tiles in load_demo():
        for _, b in tiles:
            a = TITLES.get(b, (b, None))[0]
            if a not in seen:
                seen.append(a)
    seen += [b for b in load_apps() if b not in seen and not b.startswith("com.apple.")]
    for b in seen:
        while _lock.locked():           # a launch or reset is talking to the headset: wait our turn
            time.sleep(1)
        if valid_bundle(b):
            fetch_icon(b)


def apps_rev():
    """A short fingerprint of the app list and tile file: the remote page reloads when it changes."""
    import hashlib
    h = hashlib.sha1()
    for f in (APPS_CACHE, TILES_FILE):
        try:
            h.update(open(f, "rb").read())
        except OSError:
            h.update(b"-")
    return h.hexdigest()[:12]


def apps_stale(max_age=6 * 3600):
    try:
        return not os.path.exists(APPS_CACHE) or time.time() - os.path.getmtime(APPS_CACHE) > max_age or not load_apps()
    except OSError:
        return True


def warm_up():
    """Background, for as long as the engine serves: re-read the app list when it is missing, empty
    or over six hours old (search, the not-installed marks and the guest reset all depend on it),
    then fetch icons. Checked every 30 minutes, so an engine that runs for days sees new apps."""
    while True:
        try:
            if apps_stale():
                refresh()
            warm_icons()
        except Exception as e:                      # one bad reading never ends the loop
            print(f"warm-up: {type(e).__name__}: {e}", flush=True)
        time.sleep(1800)


def launch(bundle):
    """Open one app on the headset (or one listed title inside its app). Returns (ok, message). One at a time."""
    shown_id = bundle
    bundle, url = TITLES.get(bundle, (bundle, None))
    extra = ["--payload-url", url] if url else []
    with device_turn() as mine:
        if not mine:
            return False, "Another app is opening; try again in a moment"
        ok, msg = _launch(bundle, extra)
        if ok:
            _last["id"] = shown_id               # every remote marks the tile "Last opened"
        return ok, msg


def _launch(bundle, extra):
    t0 = time.time()
    st = lock_state()
    if st != "ready":
        msg = STATE_TEXT[st]
        with open(LOG, "a") as f:
            f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {bundle} SKIP {time.time() - t0:.1f}s {st}\n")
        return False, msg
    try:
        r = subprocess.run(["xcrun", "devicectl", "device", "process", "launch", "--device", device(),
                            "--terminate-existing", "--timeout", str(TIMEOUT), *extra, bundle],
                           capture_output=True, text=True, timeout=TIMEOUT + 20)
        ok = r.returncode == 0
        msg = "opened" if ok else ((r.stderr or r.stdout).strip().splitlines() or ["failed"])[-1]
    except subprocess.TimeoutExpired:
        ok, msg = False, f"no answer from the headset in {TIMEOUT + 20} seconds (is it awake and on your head?)"
    except OSError as e:
        ok, msg = False, f"the Mac could not run Xcode's device tools ({e.strerror or e})"
    took = time.time() - t0
    os.makedirs(CFG, exist_ok=True)
    with open(LOG, "a") as f:
        f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {bundle} {'OK' if ok else 'FAIL'} {took:.1f}s {msg[:160]}\n")
    return ok, msg


def launch_stats(day=None, path=None):
    """Today's demo numbers from launches.log: how many apps opened, failed or were skipped, the
    median open time, and resets. Counts and timings only."""
    day = day or time.strftime("%Y-%m-%d")
    opened, failed, skipped, resets, took = 0, 0, 0, 0, []
    try:
        for line in open(path or LOG):
            parts = line.split()
            if len(parts) < 4 or parts[0] != day:
                continue
            if parts[2] == "reset":
                resets += 1
            elif parts[3] == "OK":
                opened += 1
                try:
                    took.append(float(parts[4].rstrip("s")))
                except (IndexError, ValueError):
                    pass
            elif parts[3] == "FAIL":
                failed += 1
            elif parts[3] == "SKIP":
                skipped += 1
    except OSError:
        pass
    import statistics
    med = round(statistics.median(took), 1) if took else None
    return {"opened": opened, "failed": failed, "skipped": skipped, "resets": resets, "median_s": med}


def reset_roots():
    """Install folders whose running processes a reset closes: every third-party app, every Apple
    app that is a demo tile (TV, Spatial Gallery, Encounter Dinosaurs, Mindfulness), and Photos,
    which shows the owner's own library. None when the app list is missing."""
    try:
        apps = json.load(open(APPS_CACHE)).get("result", {}).get("apps", [])
    except Exception:
        return None
    demo = {TITLES.get(b, (b, None))[0] for _, tiles in load_demo() for _, b in tiles} | {"com.apple.mobileslideshow"}
    roots = []
    for a in apps:
        b, u = a.get("bundleIdentifier", ""), a.get("url", "")
        if u and (not b.startswith("com.apple.") or b in demo):
            roots.append(u if u.endswith("/") else u + "/")
    return roots


CONTAINERS = "file:///private/var/containers/Bundle/Application/"


def apple_keep_names():
    """App folder names (MobileMail.app) of Apple's own App Store-style apps that a reset leaves
    running: every com.apple app in the list that is not a demo tile and not Photos."""
    try:
        apps = json.load(open(APPS_CACHE)).get("result", {}).get("apps", [])
    except Exception:
        return set()
    demo = {TITLES.get(b, (b, None))[0] for _, tiles in load_demo() for _, b in tiles} | {"com.apple.mobileslideshow"}
    return {a.get("url", "").rstrip("/").split("/")[-1] for a in apps
            if a.get("bundleIdentifier", "").startswith("com.apple.") and a.get("bundleIdentifier") not in demo and a.get("url")}


def reset_targets(procs, roots, keep=None):
    """(pid, executable) for each running app process under one of `roots`; extensions are left alone.
    With `keep` (Apple app folder names to leave running), any app running from the App Store
    install area is a target too, so an app installed or updated after the list was read (it moves
    to a new install folder) is still closed."""
    out = []
    for pr in procs:
        exe, pid = pr.get("executable", ""), pr.get("processIdentifier")
        if not pid or ".appex/" in exe:
            continue
        hit = any(exe.startswith(rt) for rt in roots)
        if not hit and keep is not None and exe.startswith(CONTAINERS):
            folder = exe[len(CONTAINERS):].split("/")[1] if exe[len(CONTAINERS):].count("/") >= 2 else ""
            hit = folder.endswith(".app") and folder not in keep
        if hit:
            out.append((pid, exe))
    return out


RESET_BUDGET = 90      # seconds: every caller (page, Mac menu, CLI) waits longer than this


def _app_name(exe):
    return exe.rstrip("/").split("/")[-1].removesuffix(".app") if exe else "an app"


def running_targets(roots, budget, dev=None):
    """(pid, executable) of the guest apps running now, or None when the headset did not answer
    within `budget` seconds or answered with something that is not a process list."""
    if budget < 4:
        return None
    tmp = os.path.join(CFG, f"procs-{threading.get_ident()}.json")
    try:
        r = subprocess.run(["xcrun", "devicectl", "device", "info", "processes", "--device", dev or device(),
                            "--timeout", str(max(2, min(15, int(budget) - 2))), "--json-output", tmp],
                           capture_output=True, text=True, timeout=min(30, budget))
        if r.returncode != 0:
            return None
        procs = json.load(open(tmp)).get("result", {}).get("runningProcesses")
        if not isinstance(procs, list) or not procs:   # a live headset always runs hundreds of processes
            return None
        return targets_from(procs, roots)
    except Exception:
        return None
    finally:
        try:
            os.remove(tmp)
        except OSError:
            pass


def terminate(pid, budget, dev=None):
    try:
        t = subprocess.run(["xcrun", "devicectl", "device", "process", "terminate", "--device", dev or device(),
                            "--pid", str(pid), "--timeout", str(int(max(3, min(15, budget - 3))))],
                           capture_output=True, text=True, timeout=max(5, min(30, budget)))
        return t.returncode == 0
    except Exception:
        return False


def targets_from(procs, roots):
    """The App Store install-area sweep runs only with a real list of Apple's own apps to leave alone
    (a full headset lists dozens); a partial or broken list falls back to the known install paths."""
    keep = apple_keep_names()
    # A real list read with --include-all-apps names dozens of Apple apps (46 on Jason's headset,
    # 2026-10-18). Fewer than 20 means a partial reading, and the sweep could reach an Apple app.
    return reset_targets(procs, roots, keep if len(keep) >= 20 and "MobileMail.app" in keep else None)


def reset_for_guest():
    """Close every app a guest may have opened: each running third-party app, plus Photos (the
    owner's own library) and the demo-tile Apple apps. Returns (ok, message). Works while the
    headset is locked. Answers within RESET_BUDGET seconds with what it closed and what it did not,
    then lists the running apps again so "Ready" is only said when none of them is left."""
    with device_turn() as mine:
        if not mine:
            return False, "An app is opening; try again in a moment"
        return _reset()


def _reset():
    t0 = time.monotonic()
    left = lambda: RESET_BUDGET - (time.monotonic() - t0)
    try:
        dev = device()                         # one headset for the whole reset, chosen once
        roots = reset_roots()
        if not roots or not load_apps():
            refresh_soon()
            return False, "The headset's app list is being re-read; try Reset again in a minute"
        targets = running_targets(roots, left(), dev)
        if targets is None:
            return False, STATE_TEXT["unreachable"]
        closed, failed, skipped = 0, [], 0
        for pid, exe in targets:
            if left() < 6:
                skipped += 1
                continue
            if terminate(pid, left(), dev):
                closed += 1
            else:
                failed.append(_app_name(exe))
        still = running_targets(roots, left(), dev)
        if still:
            failed = sorted({_app_name(e) for _, e in still} | set(failed))
        with open(LOG, "a") as f:
            f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} reset closed {closed} failed {len(failed)} "
                    f"skipped {skipped} still {len(still) if still is not None else 'unchecked'} {time.monotonic() - t0:.1f}s\n")
        if still is None:
            return False, (f"Closed {closed} app{'s' if closed != 1 else ''}; the headset did not answer the final check. "
                           "Tap Reset again to confirm")
        _last["id"] = None if not failed and not skipped else _last["id"]
        if failed or skipped:
            more = f"; {skipped} not tried in time" if skipped else ""
            return False, f"Closed {closed}; still open: {', '.join(failed[:4]) or 'unknown'}{more}. Tap Reset again"
        return True, f"Ready for the next guest ({closed} app{'s' if closed != 1 else ''} closed)"
    except Exception as e:
        return False, f"Reset stopped: {type(e).__name__}. Tap Reset again"


KEY_GENERATION = "2"   # bump to rotate the master key on every Mac at the next start


def key():
    """The Mac-only master key. Written atomically with 0600 from birth, so no reader ever sees an
    empty or world-readable file. Generation 2 (2026-10-17) rotated the key that earlier builds put
    into a link and a log; old keyed links now land on the pairing page."""
    os.makedirs(CFG, exist_ok=True)
    p = os.path.join(CFG, "key")
    gen = os.path.join(CFG, "key-generation")
    cur = open(gen).read().strip() if os.path.exists(gen) else ""
    if not os.path.exists(p) or cur != KEY_GENERATION:
        tmp = p + f".{os.getpid()}.tmp"
        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w") as f:
            f.write(secrets.token_urlsafe(24))
        os.replace(tmp, p)
        with open(gen, "w") as f:
            f.write(KEY_GENERATION)
    return open(p).read().strip()


# ---- Pairing ----------------------------------------------------------------------------------
# Each phone, iPad or browser pairs ONCE with a six-digit code shown in the Mac's menu bar app
# (or `vp-demo code`). It then holds its own token in an HttpOnly cookie, so no key ever sits in a
# URL. The master key (~/.config/vp-demo/key) stays on the Mac: the menu bar app and the CLI send it
# in an X-Key header. Every POST also needs the header X-VPD, which a cross-site form cannot send.
_pair = {"code": None, "until": 0.0}
_tries = {}
_plock = threading.Lock()
PAIR_TTL = 600


def devices_path():
    return os.path.join(CFG, "devices.json")


def load_devices():
    try:
        return json.load(open(devices_path()))
    except Exception:
        return {}


_dlock = threading.RLock()


def update_devices(fn, write=None):
    """Every change to the paired list goes through here, under one lock, so a pairing, a 'last used'
    stamp and an unpair-all can never undo each other. `write(out)` decides whether the file is
    rewritten (default: always); a read that changed nothing leaves the file alone."""
    with _dlock:
        d = load_devices()
        out = fn(d)
        if write is None or write(out):
            save_devices(d)
        return out


def save_devices(d):
    os.makedirs(CFG, exist_ok=True)
    tmp = devices_path() + f".{threading.get_ident()}.tmp"
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(d, f, indent=1)
    os.replace(tmp, devices_path())


def pair_code(rotate=False):
    with _plock:
        if rotate or not _pair["code"] or time.time() > _pair["until"]:
            _pair["code"] = f"{secrets.randbelow(1000000):06d}"
            _pair["until"] = time.time() + PAIR_TTL
            _pair["wrong"] = 0
        return _pair["code"], int(_pair["until"] - time.time())


def try_pair(code, ip, label):
    """Returns (token or None, message). Five wrong codes per address per ten minutes."""
    now = time.time()
    with _plock:
        recent = [t for t in _tries.get(ip, []) if now - t < 600]
        if len(recent) >= 5:
            _tries[ip] = recent
            return None, "Too many tries. Wait ten minutes, then use the new code on the Mac."
        good = bool(_pair["code"]) and now <= _pair["until"] and secrets.compare_digest(str(code), _pair["code"])
        if not good:
            _tries[ip] = recent + [now]
            _pair["wrong"] = _pair.get("wrong", 0) + 1
            if _pair["wrong"] >= 5:                       # five wrong from anywhere: this code is spent
                _pair.update(code=None, until=0.0, wrong=0)
                return None, "Too many wrong codes. The Mac menu bar now shows a new one."
            return None, "That code does not match. Check the code in the Mac menu bar."
        _pair["wrong"] = 0
        _pair["code"] = None
        _tries.pop(ip, None)
    return mint_device(label), "Paired"


def mint_device(label):
    tok = secrets.token_urlsafe(24)
    stamp = time.strftime("%Y-%m-%d %H:%M")
    update_devices(lambda d: d.__setitem__(tok, {"label": label[:80], "paired": stamp, "last": stamp}))
    return tok


# One-time handoff links: the Mac app opens the remote in its own browser, and the QR code pairs a
# phone in one scan, without the master key ever appearing in a URL.
_handoffs = {}


_hosts = {"at": 0.0, "set": frozenset(), "names": frozenset()}
_hosts_lock = threading.Lock()


def _discover():
    """This Mac's names and addresses, read fresh. A lookup that fails returns nothing for that part."""
    names, addrs = set(), set()
    for which in ("LocalHostName", "HostName"):
        try:
            r = subprocess.run(["/usr/sbin/scutil", "--get", which], capture_output=True, text=True, timeout=3)
            n = r.stdout.strip().lower().rstrip(".") if r.returncode == 0 else ""
        except Exception:
            n = ""
        if n:
            names.add(n)                                   # the bare name works on many home networks
            if "." not in n:
                names.add(n + ".local")                    # and its Bonjour form, which the phone uses
    try:
        out = subprocess.run(["/sbin/ifconfig"], capture_output=True, text=True, timeout=3).stdout
        addrs.update(m.group(1).lower() for m in re.finditer(r"\binet6? ([0-9a-fA-F:.]+)", out))
    except Exception:
        pass
    return names, addrs


def allowed_hosts(max_age=30, discover=None):
    """The names and addresses this Mac answers to: loopback, its Bonjour and host names, and every
    interface address. One refresh runs at a time (every other request reads the current list), and
    a name lookup that fails keeps the last good names, so a hiccup never refuses the paired phone."""
    if time.time() - _hosts["at"] < max_age and _hosts["set"]:
        return _hosts["set"]
    if not _hosts_lock.acquire(blocking=not _hosts["set"]):
        return _hosts["set"]                   # another request is refreshing; use what we have
    try:
        names, addrs = (discover or _discover)()
        names = frozenset(names) or _hosts["names"]
        _hosts.update(at=time.time(), names=names,
                      set=frozenset({"localhost", "127.0.0.1", "::1"} | set(names) | set(addrs)))
        return _hosts["set"]
    finally:
        _hosts_lock.release()


def host_ok(h):
    """A request must name this Mac. A web page that points its own domain at this Mac (DNS
    rebinding) sends its own name in Host and is refused, before any route runs."""
    h = (h or "").strip().lower()
    if h.startswith("["):
        h = h[1:].split("]")[0].split("%")[0]      # drop an IPv6 zone such as %en0 or %25en0
    elif h.count(":") == 1:
        h = h.split(":")[0]
    h = h.rstrip(".")
    if not h:
        return False
    # A miss right after the Mac joins a network re-reads its addresses at once (at most every 5 s),
    # so the phone is never refused for half a minute by a stale list.
    return h in allowed_hosts() or h in allowed_hosts(max_age=5)


def engine_is_ours(port=8787):
    """True only when the process listening on the port is the engine named in serve.pid, so the
    master key is never sent to some other program that took the port."""
    try:
        pids = subprocess.run(["/usr/sbin/lsof", "-nP", f"-iTCP:{port}", "-sTCP:LISTEN", "-t"],
                              capture_output=True, text=True, timeout=5).stdout.split()
        with open(os.path.join(CFG, "serve.pid")) as f:
            mine = f.read().strip()
        if not mine.isdigit():
            return False
        cmd = subprocess.run(["/bin/ps", "-o", "command=", "-p", mine], capture_output=True, text=True, timeout=5).stdout.split()
    except Exception:
        return False
    # Only our engine may be listening: a second listener on the same port (another address, say)
    # makes the destination ambiguous, so the key stays home. Any program running as this user can
    # read the key file anyway; this guards against a port taken by mistake, not against that.
    return pids == [mine] and any(w.endswith("vp-demo") for w in cmd) and "serve" in cmd


def new_handoff(ttl):
    t = secrets.token_urlsafe(18)
    with _plock:
        now = time.time()
        for k_ in [k_ for k_, v_ in _handoffs.items() if v_ < now]:
            _handoffs.pop(k_, None)
        _handoffs[t] = now + ttl
    return t


def take_handoff(t):
    with _plock:
        until = _handoffs.pop(t, 0)
    return bool(t) and until >= time.time()


def device_label(ua):
    ua = ua or ""
    for k, v in (("iPad", "iPad"), ("iPhone", "iPhone"), ("Macintosh", "Mac"), ("Android", "Android")):
        if k in ua:
            return v
    return "Browser"


APP_NAME = "Vision Test Drive"
SHORT_NAME = "Test Drive"   # the Home Screen label: iOS cuts a label near 12 characters
ICON_SVG = """<svg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 1024 1024'>
<defs><linearGradient id='g' x1='0' y1='0' x2='0' y2='1'><stop offset='0' stop-color='#33383D'/><stop offset='1' stop-color='#191B1D'/></linearGradient></defs>
<rect width='1024' height='1024' rx='228' fill='url(#g)'/>
<path d='M344 318H680Q792 318 792 430V514Q792 640 666 640H620Q584 640 566 606L556 586Q540 556 512 556Q484 556 468 586L458 606Q440 640 404 640H358Q232 640 232 514V430Q232 318 344 318Z' fill='none' stroke='#C8A872' stroke-width='40' stroke-linejoin='round'/>
<circle cx='512' cy='760' r='30' fill='#F1EBDD'/><circle cx='512' cy='760' r='70' fill='none' stroke='#F1EBDD' stroke-opacity='.45' stroke-width='16'/>
</svg>"""


def app_icon_png(size):
    """The remote's own icon as a PNG (for the Home Screen), rendered once and cached."""
    bundled = os.path.join(HERE, f"app-icon-{size}.png")      # build.sh puts these beside the engine
    if os.path.exists(bundled) and os.path.getsize(bundled) > 0:
        return bundled
    os.makedirs(CFG, exist_ok=True)
    out = os.path.join(CFG, f"app-icon-{size}.png")
    if not os.path.exists(out):
        svg = os.path.join(CFG, "app-icon.svg")
        open(svg, "w").write(ICON_SVG)
        for tool in ("/opt/homebrew/bin/rsvg-convert", "/usr/local/bin/rsvg-convert"):
            if os.path.exists(tool):
                subprocess.run([tool, "-w", str(size), "-h", str(size), "-o", out, svg], capture_output=True)
                break
    return out if os.path.exists(out) else None


def manifest():
    return json.dumps({"name": APP_NAME, "short_name": SHORT_NAME, "start_url": "/", "scope": "/", "display": "standalone",
                       "background_color": "#141412", "theme_color": "#141412",
                       "icons": [{"src": "/app-icon-512.png", "sizes": "512x512", "type": "image/png"},
                                 {"src": "/app-icon-180.png", "sizes": "180x180", "type": "image/png"}]})


HEAD = ("<!doctype html><html lang=en><meta charset=utf-8><meta name=viewport content='width=device-width,initial-scale=1,viewport-fit=cover'>"
        "<meta name=apple-mobile-web-app-capable content=yes><meta name=mobile-web-app-capable content=yes>"
        "<meta name=apple-mobile-web-app-status-bar-style content=black-translucent>"
        f"<meta name=apple-mobile-web-app-title content='{SHORT_NAME}'><link rel=manifest href=/manifest.webmanifest>"
        "<link rel=apple-touch-icon href=/app-icon-180.png><meta name=theme-color content='#141412'>")


def pair_page(msg=""):
    return (HEAD + f"<title>{APP_NAME}: pair</title><style>{PAGE_CSS}</style>"
            f"<h1>Pair this device</h1><p class=lead>Open <b>{APP_NAME}</b> in the Mac's menu bar and type the six-digit code shown there.</p>"
            "<form id=f><label for=c class=sr>Pairing code</label><input id=c inputmode=numeric autocomplete=one-time-code maxlength=7 placeholder='123 456' autofocus>"
            "<button class=go id=go type=submit>Pair</button></form>"
            f"<div id=status role=status aria-live=polite>{html.escape(msg)}</div>"
            "<p class=lead>Tip: in Safari, tap Share, then Add to Home Screen, then pair from the Home Screen icon.</p>"
            """<script>const S=document.getElementById('status'),C=document.getElementById('c'),G=document.getElementById('go');
C.oninput=()=>{const d=C.value.replace(/\D/g,'').slice(0,6);if(d.length===6&&!G.disabled)document.getElementById('f').requestSubmit();};
document.getElementById('f').onsubmit=async e=>{e.preventDefault();const code=C.value.replace(/\D/g,'');
 if(code.length!==6){S.textContent='Type all six digits from the Mac menu bar.';return;}
 G.disabled=true;S.textContent='Pairing...';const ac=new AbortController(),to=setTimeout(()=>ac.abort(),20000);
 try{const r=await fetch('/pair',{method:'POST',headers:{'X-VPD':'1','Content-Type':'application/json'},body:JSON.stringify({code}),signal:ac.signal});
  const j=await r.json();if(j.ok){location.replace('/');return;}S.textContent=j.message;C.select();}
 catch(err){S.textContent=err.name==='AbortError'?'No answer yet from the Mac. If this device paired, reloading opens the remote; otherwise try again.':'The Mac did not answer. Check this device is on the same Wi-Fi as the Mac, then try again.';}
 finally{clearTimeout(to);G.disabled=false;}};</script>""")


PAGE_CSS = """
:root{--bg:#f6f4ef;--card:#fff;--ink:#1c1b19;--muted:#6b675f;--line:#e4e0d6;--gold:#9a7b3c;--goldtext:#86622a;--ok:#2f6b3a;--bad:#9b2c2c}
@media (prefers-color-scheme:dark){:root{--bg:#141412;--card:#1f1e1b;--ink:#f1eee7;--muted:#a39e93;--line:#34322d;--gold:#d4b46a;--goldtext:#d4b46a;--ok:#7fc58b;--bad:#ef8b8b}}
*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--ink);font:17px/1.4 -apple-system,system-ui,sans-serif;padding:16px}
h1{font-size:24px;margin:8px 0 4px}h2{font-size:15px;letter-spacing:.06em;text-transform:uppercase;color:var(--muted);margin:24px 0 8px}
.grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(150px,1fr));gap:12px}
button.tile{background:var(--card);border:1px solid var(--line);border-radius:16px;padding:18px 14px;min-height:84px;
 font:inherit;font-weight:600;color:var(--ink);text-align:left;cursor:pointer}
button.tile:active{transform:scale(.98)}button:focus-visible,input:focus-visible{outline:3px solid var(--gold);outline-offset:2px}
button.tile:disabled{cursor:default;opacity:.55}button.tile[aria-busy=true]:not(.busy){opacity:.6;cursor:default}.sr{position:absolute;width:1px;height:1px;overflow:hidden;clip:rect(0 0 0 0)}
#none{color:var(--muted);display:none}
.lastl{display:block;font-size:13px;font-weight:600;color:var(--ok);margin-top:4px}button.tile.last{border-color:var(--ok)}button.tile.busy{opacity:.6}button.tile.ok{border-color:var(--ok)}button.tile.bad{border-color:var(--bad)}
.missing{color:var(--muted);font-weight:400;font-size:13px;display:block}
button.tile img{width:52px;height:52px;border-radius:50%;display:block;margin-bottom:10px}
.pill.noxcode{color:var(--bad);border-color:var(--bad)}.pill{display:inline-block;font-size:14px;padding:4px 10px;border-radius:999px;border:1px solid var(--line);color:var(--muted)}
.pill.ready{color:var(--ok);border-color:var(--ok)}.pill.locked{color:var(--goldtext);border-color:var(--gold)}.pill.unreachable{color:var(--bad);border-color:var(--bad)}
#status{position:sticky;top:0;background:var(--bg);padding:8px 0;color:var(--muted);min-height:2.4em}
input{width:100%;padding:12px 14px;border-radius:12px;border:1px solid var(--line);background:var(--card);color:var(--ink);font:inherit}
body{padding:max(16px,env(safe-area-inset-top)) 16px 32px}.lead{color:var(--muted)}
#c{font-size:28px;letter-spacing:.3em;text-align:center;margin:12px 0}
button.go,button.reset{width:100%;padding:14px;border-radius:14px;border:0;background:var(--gold);color:#141412;font:inherit;font-weight:700}
button.reset{background:var(--card);color:var(--ink);border:1px solid var(--line);margin-top:28px}
"""


PAGE_JS = """<script>
const S=document.getElementById('status'),H={'X-VPD':'1'},R=document.getElementById('reset');let busy=false,rev=REV0,reloadDue=false;
async function post(u,ms){const c=new AbortController(),t=setTimeout(()=>c.abort(),ms);
 try{const r=await fetch(u,{method:'POST',headers:H,signal:c.signal});if(r.status===401){location.replace('/pair');throw new Error('not paired');}return await r.json();}
 finally{clearTimeout(t);}}
function lost(e,what){return e.name==='AbortError'?'No answer yet from the Mac; '+what+' may still finish. Wait a moment before trying again.':'The Mac did not answer. Check this device is on the same Wi-Fi as the Mac.';}
function hold(on){if(!on&&reloadDue){location.reload();return;}busy=on;R.disabled=on;document.querySelectorAll('button.tile').forEach(b=>b.setAttribute('aria-busy',on?'true':'false'));}
R.onclick=async()=>{if(busy)return;hold(true);S.textContent="Closing the last guest's apps...";
 try{const j=await post('/reset',120000);S.textContent=j.message;if(j.ok)markLast(null);}catch(e){S.textContent=lost(e,'the reset');}hold(false);};
document.querySelectorAll('button.tile').forEach(t=>t.onclick=async()=>{
 if(busy||t.disabled)return;const nm=t.querySelector('.n').textContent;hold(true);t.classList.remove('ok','bad');t.classList.add('busy');S.textContent='Opening '+nm+'...';
 try{const j=await post('/launch?b='+encodeURIComponent(t.dataset.b),100000);t.classList.add(j.ok?'ok':'bad');if(j.ok)markLast(t.dataset.b);S.textContent=(j.ok?'Opened ':'Could not open ')+nm+(j.ok?'':': '+j.message);if(navigator.vibrate)navigator.vibrate(j.ok?20:[30,40,30]);}
 catch(e){t.classList.add('bad');S.textContent=lost(e,nm);}
 t.classList.remove('busy');hold(false);});
const P=document.getElementById('pill');async function poll(){if(document.hidden)return;try{const r=await fetch('/status');if(r.status===401){location.replace('/pair');return;}const j=await r.json();
 if(P.textContent!==j.text)P.textContent=j.text;P.className='pill '+j.state;markLast(j.last);
 if(j.rev&&j.rev!==rev){if(busy){reloadDue=true;}else{location.reload();return;}}}catch(e){P.textContent='The Mac did not answer';P.className='pill unreachable';}}
function markLast(id){document.querySelectorAll('button.tile').forEach(b=>{const on=!!id&&b.dataset.b===id;b.classList.toggle('last',on);
 let l=b.querySelector('.lastl');if(on&&!l){l=document.createElement('span');l.className='lastl';l.textContent='Last opened';b.appendChild(l);}if(!on&&l)l.remove();});}
poll();setInterval(poll,15000);document.addEventListener('visibilitychange',()=>{if(!document.hidden)poll();});
const q=document.getElementById('q'),N=document.getElementById('none');if(q)q.oninput=()=>{const v=q.value.toLowerCase();let n=0;
 document.querySelectorAll('#all button').forEach(b=>{const m=b.dataset.n.includes(v);b.style.display=m?'':'none';n+=m?1:0;});N.style.display=n?'none':'block';};
</script>"""


def page(k=None):
    apps = load_apps()
    out = [HEAD + f"<title>{APP_NAME}</title><style>{PAGE_CSS}</style>",
           f"<h1>{APP_NAME}</h1><span id=pill class=pill role=status>Checking the headset...</span><div id=status role=status aria-live=polite>Tap an experience to open it on the headset.</div>"]
    shown = set()
    for section, tiles in load_demo():
        out.append(f"<h2>{html.escape(section)}</h2><div class=grid>")
        for n, b in tiles:
            shown.add(b)
            app = TITLES.get(b, (b, None))[0]
            miss = "" if (not apps or app in apps) else "<span class=missing>not installed on this headset</span>"
            out.append(f"<button class=tile{' disabled' if miss else ''} data-b='{html.escape(b)}'><img src='/icon?b={urllib.parse.quote(app)}' alt='' onerror='this.remove()'><span class=n>{html.escape(n)}</span>{miss}</button>")
        out.append("</div>")
    rest = sorted((n, b) for b, n in apps.items() if b not in shown and not b.startswith("com.apple."))
    if rest:
        out.append("<h2>Every other app on the headset</h2><label for=q class=sr>Search apps</label><input id=q type=search placeholder='Search apps' autocomplete=off><div class=grid id=all>")
        for n, b in rest:
            out.append(f"<button class=tile data-b='{html.escape(b)}' data-n='{html.escape(n.lower())}'><img src='/icon?b={urllib.parse.quote(b)}' alt='' loading=lazy onerror='this.remove()'><span class=n>{html.escape(n)}</span></button>")
        out.append("</div><p id=none>No apps match.</p>")
    out.append("<button class=reset id=reset>Reset for the next guest</button>")
    out.append(f"<script>const REV0='{apps_rev()}';</script>")    # the list this page was drawn from
    out.append(PAGE_JS)
    return "".join(out)


class H(BaseHTTPRequestHandler):
    timeout = 15        # a client that opens a connection and stalls is dropped, not waited on

    def _q(self):
        return urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)

    def _cookie(self):
        for part in (self.headers.get("Cookie") or "").split(";"):
            n, _, v = part.strip().partition("=")
            if n == "vpd":
                return v
        return ""

    def _auth(self):
        """'master' (X-Key header: the menu bar app or CLI), 'device' (paired cookie), or None."""
        hk = self.headers.get("X-Key") or ""
        if hk and secrets.compare_digest(hk, key()):
            return "master"
        tok = self._cookie()
        if tok:
            stamp = time.strftime("%Y-%m-%d %H:%M")

            def touch(d):
                # (paired, changed): an unknown cookie is never written; a known one only once a minute
                if tok not in d:
                    return False, False
                changed = d[tok].get("last") != stamp
                d[tok]["last"] = stamp
                return True, changed
            if update_devices(touch, write=lambda r: r[1])[0]:
                return "device"
        return None

    def _send(self, code, body, ctype="application/json", extra=None):
        b = body.encode() if isinstance(body, str) else body
        self.send_response(code)
        self.send_header("Content-Type", ctype + ("" if ctype.startswith("image/") else "; charset=utf-8"))
        self.send_header("Cache-Control", "no-store")
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Referrer-Policy", "no-referrer")
        for k_, v_ in (extra or {}).items():
            self.send_header(k_, v_)
        self.send_header("Content-Length", str(len(b)))
        self.end_headers()
        self.wfile.write(b)

    def _cookie_header(self, tok):
        return f"vpd={tok}; Path=/; Max-Age=31536000; HttpOnly; SameSite=Strict"

    def do_GET(self):
        if not host_ok(self.headers.get("Host")):
            return self._send(421, json.dumps({"ok": False, "message": "this server answers only to this Mac's own names"}))
        path = urllib.parse.urlparse(self.path).path
        q = self._q()
        if path == "/manifest.webmanifest":
            return self._send(200, manifest(), "application/manifest+json")
        if path in ("/app-icon-180.png", "/app-icon-512.png", "/apple-touch-icon.png"):
            f = app_icon_png(512 if "512" in path else 180)
            return self._send(200, open(f, "rb").read(), "image/png") if f else self._send(404, "no icon", "text/plain")
        if path == "/pair":
            t = q.get("t", [""])[0]
            if t and take_handoff(t):
                tok = mint_device(device_label(self.headers.get("User-Agent")))
                return self._send(302, "", "text/plain", {"Location": "/", "Set-Cookie": self._cookie_header(tok)})
            if self._auth():
                return self._send(302, "", "text/plain", {"Location": "/"})
            return self._send(200, pair_page("That link was already used or has expired. Type the code instead." if t else ""), "text/html")
        who = self._auth()
        if not who:
            if path == "/":
                return self._send(302, "", "text/plain", {"Location": "/pair"})
            return self._send(401, json.dumps({"ok": False, "message": "not paired"}))
        if path == "/status":
            st = cached_lock_state()
            return self._send(200, json.dumps({"state": st, "text": STATE_TEXT[st], "last": _last["id"], "rev": apps_rev()}))
        if path == "/icon":
            b = q.get("b", [""])[0]
            f = os.path.join(ICONS, b + ".png") if valid_bundle(b) and b not in TITLES else None
            if not f or not os.path.exists(f):
                return self._send(404, "no icon", "text/plain")
            return self._send(200, open(f, "rb").read(), "image/png", {"Cache-Control": "max-age=86400"})
        if path == "/catalog":
            apps = load_apps()
            return self._send(200, json.dumps({"sections": [{"name": sname, "tiles": [{"name": n, "id": b,
                "installed": (not apps) or TITLES.get(b, (b, None))[0] in apps} for n, b in tiles]} for sname, tiles in load_demo()]}))
        if path == "/admin/code" and who == "master":
            c, left = pair_code()
            return self._send(200, json.dumps({"code": c, "seconds_left": left}))
        if path == "/admin/handoff" and who == "master":
            t = q.get("t", [""])[0]
            with _plock:
                alive = _handoffs.get(t, 0) >= time.time()
            return self._send(200, json.dumps({"alive": alive}))
        if path == "/admin/stats" and who == "master":
            return self._send(200, json.dumps(launch_stats()))
        if path == "/admin/devices" and who == "master":
            return self._send(200, json.dumps([{"id": t[:6], **v} for t, v in load_devices().items()]))
        if path == "/":
            return self._send(200, page(), "text/html")
        return self._send(404, "not found", "text/plain")

    def do_POST(self):
        if not host_ok(self.headers.get("Host")):
            return self._send(421, json.dumps({"ok": False, "message": "this server answers only to this Mac's own names"}))
        path = urllib.parse.urlparse(self.path).path
        q = self._q()
        if self.headers.get("X-VPD") != "1" and not self.headers.get("X-Key"):
            # A cross-site form cannot set a custom header; a fetch from another origin that tries is
            # preflighted, and this server never answers a preflight.
            return self._send(403, json.dumps({"ok": False, "message": "missing X-VPD header"}))
        if path == "/pair":
            try:
                n = min(int(self.headers.get("Content-Length") or 0), 512)
                code = (json.loads(self.rfile.read(n) or b"{}").get("code") or "")
            except Exception:
                code = ""
            code = re.sub(r"\D", "", str(code))            # "123 456" as the menu shows it is the same code
            tok, msg = try_pair(code, self.client_address[0], device_label(self.headers.get("User-Agent")))
            if not tok:
                return self._send(403, json.dumps({"ok": False, "message": msg}))
            return self._send(200, json.dumps({"ok": True, "message": msg}), extra={"Set-Cookie": self._cookie_header(tok)})
        who = self._auth()
        if not who:
            return self._send(401, json.dumps({"ok": False, "message": "not paired"}))
        if path == "/launch":
            b = q.get("b", [""])[0]
            if not valid_bundle(b):
                return self._send(400, json.dumps({"ok": False, "message": "bad bundle id"}))
            if not allowed(b):
                return self._send(403, json.dumps({"ok": False, "message": "that app is not a demo tile or installed on the headset"}))
            good, msg = launch(b)
            return self._send(200, json.dumps({"ok": good, "message": msg}))
        if path == "/reset":
            good, msg = reset_for_guest()
            return self._send(200, json.dumps({"ok": good, "message": msg}))
        if path == "/admin/rotate" and who == "master":
            c, left = pair_code(rotate=True)
            return self._send(200, json.dumps({"code": c, "seconds_left": left}))
        if path == "/admin/handoff" and who == "master":
            ttl = 600 if q.get("for", [""])[0] == "qr" else 60
            c, left = pair_code()              # the QR window shows both, so it gets both in one answer
            return self._send(200, json.dumps({"t": new_handoff(ttl), "seconds": ttl, "code": c, "code_seconds": left}))
        if path == "/admin/refresh" and who == "master":
            refresh_soon()
            return self._send(200, json.dumps({"ok": True, "message": "Re-reading the headset's apps"}))
        if path == "/admin/unpair-all" and who == "master":
            update_devices(lambda d: d.clear())
            return self._send(200, json.dumps({"ok": True, "message": "Every device unpaired"}))
        return self._send(404, json.dumps({"ok": False, "message": "not found"}))

    def log_message(self, *a):
        pass


def selftest():
    fails = 0
    k = "selftest-key"
    # Decoy: bundle ids carrying shell or path characters must be refused by the same check the server uses.
    for b in ["a;rm", "x y", "com.a|b", "$(id)", "../x", "", "a/b"]:
        if valid_bundle(b):
            print(f"FAIL  decoy: '{b}' would reach devicectl"); fails += 1
    if not fails:
        print("PASS  decoy: bundle ids with shell or path characters are refused")
    if valid_bundle("title:nope") or not valid_bundle("title:elevated"):
        print("FAIL  decoy: an unlisted title key was accepted, or a listed one refused"); fails += 1
    else:
        print("PASS  decoy: only listed titles are accepted")
    if not valid_bundle("com.VirZOOM.FlyVP") or not valid_bundle("com.stayhvn.havenvision"):
        print("FAIL  control: a real bundle id was refused"); fails += 1
    # Control: a real bundle id passes the filter and resolves by display name.
    apps = {"com.VirZOOM.FlyVP": "FLY"}
    if resolve("fly", apps) != "com.VirZOOM.FlyVP" or resolve("FLY", apps) != "com.VirZOOM.FlyVP":
        print("FAIL  control: 'fly' did not resolve to com.VirZOOM.FlyVP"); fails += 1
    else:
        print("PASS  control: names resolve to bundle ids")
    # The owner's tile file: honored when good, a bad id inside it dropped, a broken file falls back.
    import tempfile as _tf
    td = _tf.mkdtemp(prefix="vp-demo-tiles-")
    good, bad = os.path.join(td, "good.json"), os.path.join(td, "bad.json")
    json.dump([["Mine", [["Mine A", "com.example.mine"], ["Sneaky", "a;rm -rf"]]]], open(good, "w"))
    open(bad, "w").write("{not json")
    mine = load_demo(good)
    if mine != [("Mine", [("Mine A", "com.example.mine")])]:
        print(f"FAIL  control: the owner's tiles.json was not honored with its bad id dropped: {mine}"); fails += 1
    else:
        print("PASS  control: the owner's tiles.json is honored, and an unsafe id inside it is dropped")
    if load_demo(bad) is not DEFAULT_DEMO or load_demo(os.path.join(td, "absent.json")) is not DEFAULT_DEMO:
        print("FAIL  decoy: a broken or missing tiles.json did not fall back to the public default"); fails += 1
    else:
        print("PASS  decoy: a broken or missing tiles.json falls back to the public default")
    flat = " ".join(b for _, t in DEFAULT_DEMO for _, b in t)
    if any(w in flat for w in ("stayhvn", "partridgepaez", "jasonpaez", "epiphani")):
        print("FAIL  decoy: the public default carries a private or studio-internal app"); fails += 1
    else:
        print("PASS  decoy: the public default carries only public apps")
    # Headset discovery: simulators are skipped, the first physical visionOS device wins.
    listing = {"result": {"devices": [
        {"identifier": "SIM", "hardwareProperties": {"platform": "visionOS", "reality": "simulated"}},
        {"identifier": "PHONE", "hardwareProperties": {"platform": "iOS", "reality": "physical"}},
        {"identifier": "REAL", "hardwareProperties": {"platform": "visionOS", "reality": "physical"}}]}}
    if discover_device(listing) != "REAL" or discover_device({"result": {"devices": listing["result"]["devices"][:2]}}) is not None:
        print("FAIL  control: headset discovery picked a simulator or a phone"); fails += 1
    else:
        print("PASS  control: headset discovery skips simulators and phones")
    if "button class=tile" not in page(k):
        print("FAIL  control: the page has no tiles"); fails += 1
    else:
        print("PASS  control: the page renders its tiles")
    for label, doc in (("remote page", page(k)), ("pair page", pair_page())):
        err = js_syntax_error(doc)
        if err == "NOT MEASURED":
            print(f"NOT MEASURED  {label}: no JavaScriptCore runner (osascript) on this Mac")
        elif err:
            print(f"FAIL  {label}: its script does not parse in JavaScriptCore: {err}"); fails += 1
        else:
            print(f"PASS  {label}: its script parses in JavaScriptCore (Safari's engine)")
    if not js_syntax_error("<script>const a='it's';</script>"):
        print("FAIL  decoy: the JavaScript check passed a broken script"); fails += 1
    else:
        print("PASS  decoy: the JavaScript check catches a stray quote")
    procs = [{"processIdentifier": 1, "executable": "file:///Applications/TV.app/TV"},
             {"processIdentifier": 2, "executable": "file:///private/var/containers/Bundle/Application/A/FLY.app/FLY"},
             {"processIdentifier": 3, "executable": "file:///Applications/RealityLauncher.app/RealityLauncher"},
             {"processIdentifier": 4, "executable": "file:///Applications/TV.app/PlugIns/X.appex/X"}]
    roots = ["file:///Applications/TV.app/", "file:///private/var/containers/Bundle/Application/A/FLY.app/"]
    got = sorted(pid for pid, _ in reset_targets(procs, roots))
    if got != [1, 2]:
        print(f"FAIL  reset: expected TV and FLY closed and the Home View and extensions left alone, got {got}"); fails += 1
    else:
        print("PASS  reset: closes TV and a third-party app, leaves the Home View and extensions alone")
    # Lock state: a reading without passcodeRequired is unknown, never "ready".
    if parse_lock({}) != "unreachable" or parse_lock({"unlockedSinceBoot": True}) != "unreachable" \
            or parse_lock({"passcodeRequired": False}) != "ready" or parse_lock({"passcodeRequired": True}) != "locked":
        print("FAIL  decoy: a lock reading without passcodeRequired was read as ready"); fails += 1
    else:
        print("PASS  decoy: a lock reading without passcodeRequired reads unknown; a real one reads ready or locked")
    two = {"result": {"devices": [{"identifier": "A", "hardwareProperties": {"platform": "visionOS", "reality": "physical"}},
                                  {"identifier": "B", "hardwareProperties": {"platform": "visionOS", "reality": "physical"}}]}}
    if discover_device(two, single=True) is not None or discover_device(listing, single=True) != "REAL":
        print("FAIL  decoy: re-discovery picked one of two headsets on its own"); fails += 1
    else:
        print("PASS  decoy: re-discovery leaves the saved headset alone when two are listed, and takes the only one")
    procs2 = [{"processIdentifier": 5, "executable": CONTAINERS + "NEW-UUID/FLY.app/FLY"},
              {"processIdentifier": 6, "executable": CONTAINERS + "U2/MobileMail.app/MobileMail"},
              {"processIdentifier": 8, "executable": CONTAINERS + "U3/FLY.app/PlugIns/W.appex/W"}]
    got2 = sorted(p_ for p_, _ in reset_targets(procs2, [], {"MobileMail.app"}))
    got3 = sorted(p_ for p_, _ in reset_targets(procs2, [], None))
    if got3 != []:
        print(f"FAIL  decoy: with no list of Apple's apps to keep, Reset reached into the App Store install area: {got3}"); fails += 1
    else:
        print("PASS  decoy: without a full list of Apple's apps, Reset closes only apps it knows (Mail is safe)")
    if got2 != [5]:
        print(f"FAIL  reset: an app updated since the list was read should close and Mail stay, got {got2}"); fails += 1
    else:
        print("PASS  reset: an app updated since the list was read still closes; Apple's Mail and extensions stay")
    if headsets(two) != [("A", "A"), ("B", "B")] or headsets({"result": {"devices": []}}) != []:
        print("FAIL  control: the headset list"); fails += 1
    else:
        print("PASS  control: `vp-demo headsets` lists each physical headset")
    import tempfile as _tf2
    lg = os.path.join(_tf2.mkdtemp(prefix="vp-demo-log-"), "l.log")
    open(lg, "w").write("2026-10-17 10:00:00 com.a OK 2.0s opened\n2026-10-17 10:01:00 com.b OK 4.0s opened\n"
                        "2026-10-17 10:02:00 com.c FAIL 1.0s x\n2026-10-17 10:03:00 reset closed 2 failed 0\n"
                        "2026-10-16 10:00:00 com.a OK 9.0s opened\n")
    st_ = launch_stats("2026-10-17", lg)
    if st_ != {"opened": 2, "failed": 1, "skipped": 0, "resets": 1, "median_s": 3.0}:
        print(f"FAIL  control: today's numbers from the launch log: {st_}"); fails += 1
    else:
        print("PASS  control: today's numbers count only today's lines (2 opened, 1 failed, 1 reset, median 3.0 s)")
    import tempfile as _tf3
    _g = globals(); _sv = {n: _g[n] for n in ("CFG", "discover_device")}
    _sv["headsets"] = _g["headsets"]
    _g.update(CFG=_tf3.mkdtemp(prefix="vp-demo-dev-"), discover_device=lambda listing=None, single=False: None if single else "A",
              headsets=lambda listing=None: [("A", "Office"), ("B", "Den")])
    _env = os.environ.pop("VP_DEMO_DEVICE", None)
    try:
        picked = device()
    finally:
        _g.update(_sv)
        if _env is not None:
            os.environ["VP_DEMO_DEVICE"] = _env
    if picked != "no-vision-pro-paired":
        print(f"FAIL  decoy: with two headsets paired, the engine chose {picked} on its own"); fails += 1
    else:
        print("PASS  decoy: with two headsets paired, the engine waits for `vp-demo use` instead of choosing")
    _g = globals(); _sv = {n: _g[n] for n in ("CFG", "headsets")}
    _g["CFG"] = _tf3.mkdtemp(prefix="vp-demo-red-")
    try:
        save_device("B", "Living Room")
        _g["headsets"] = lambda listing=None: [("A", "Office")]
        rediscover()
        kept = open(os.path.join(CFG, "device")).read().strip() == "B"
        _g["headsets"] = lambda listing=None: [("B2", "Living Room")]
        rediscover()
        followed = open(os.path.join(CFG, "device")).read().strip() == "B2"
        os.remove(os.path.join(CFG, "device-name"))                 # an older install: id only
        _g["headsets"] = lambda listing=None: [("A", "Office")]
        rediscover()
        kept = kept and open(os.path.join(CFG, "device")).read().strip() == "B2"
    finally:
        _g.update(_sv)
    if kept and followed:
        print("PASS  re-discovery follows the chosen headset to a new id and never swaps to a different one")
    else:
        print(f"FAIL  re-discovery: kept the chosen headset {kept}, followed its new id {followed}"); fails += 1
    fails += lock_selftest()
    fails += reset_selftest()
    fails += http_selftest()
    return 1 if fails else 0


def lock_selftest():
    """A second process holding the device lock makes a launch wait its turn instead of overlapping."""
    import tempfile
    g = globals()
    saved = g["CFG"]
    g["CFG"] = tempfile.mkdtemp(prefix="vp-demo-lock-")
    try:
        holder = subprocess.Popen([sys.executable, "-c", "import fcntl,os,sys,time;fd=os.open(sys.argv[1],os.O_RDWR|os.O_CREAT);"
                                   "fcntl.flock(fd,fcntl.LOCK_EX);print('held',flush=True);time.sleep(5)",
                                   os.path.join(g["CFG"], "device.lock")], stdout=subprocess.PIPE, text=True)
        holder.stdout.readline()
        with device_turn() as mine:
            blocked = not mine
        holder.kill(); holder.wait()
        with device_turn() as mine:
            free = mine
        ok = blocked and free
        print(("PASS  " if ok else "FAIL  ") + "decoy: another process holding the headset's turn blocks a launch; released, the turn is free")
        return 0 if ok else 1
    finally:
        g["CFG"] = saved


def reset_selftest():
    """Reset with stand-ins for the headset: an empty app list is refused, a closed app reads Ready,
    an app still running after the close is named."""
    import tempfile
    g, fails = globals(), 0
    saved = {n: g[n] for n in ("CFG", "APPS_CACHE", "LOG", "running_targets", "terminate", "refresh_soon", "device")}
    tmp = tempfile.mkdtemp(prefix="vp-demo-reset-")
    asked = []
    try:
        g.update(CFG=tmp, APPS_CACHE=os.path.join(tmp, "apps.json"), LOG=os.path.join(tmp, "launches.log"),
                 refresh_soon=lambda: asked.append(1), terminate=lambda pid, b, dev=None: True, device=lambda: "HEADSET-A")
        json.dump({"result": {"apps": []}}, open(g["APPS_CACHE"], "w"))
        ok, msg = reset_for_guest()
        cond = not ok and asked and "re-read" in msg
        print(("PASS  " if cond else "FAIL  ") + "decoy: Reset with an empty app list starts a re-read and does not say Ready"); fails += 0 if cond else 1
        json.dump({"result": {"apps": [{"bundleIdentifier": "com.VirZOOM.FlyVP", "url": "file:///A/FLY.app/"}]}}, open(g["APPS_CACHE"], "w"))
        calls = []
        def rt(roots, budget, dev=None):
            calls.append(1)
            return [(7, "file:///A/FLY.app/FLY")] if len(calls) == 1 else []
        g["running_targets"] = rt
        ok, msg = reset_for_guest()
        cond = ok and "Ready" in msg and len(calls) == 2
        print(("PASS  " if cond else "FAIL  ") + "control: Reset closes the app, checks again, and says Ready"); fails += 0 if cond else 1
        calls2 = []
        def rt2(roots, budget, dev=None):
            calls2.append(1)
            return [(7, "file:///A/FLY.app/FLY")] if len(calls2) == 1 else None
        g["running_targets"] = rt2
        _last["id"] = "com.VirZOOM.FlyVP"
        ok, msg = reset_for_guest()
        cond = not ok and "final check" in msg and _last["id"] == "com.VirZOOM.FlyVP"
        print(("PASS  " if cond else "FAIL  ") + "decoy: a final check that gets no answer is never Ready, and Last opened stays"); fails += 0 if cond else 1
        procs_mail = [{"processIdentifier": 9, "executable": CONTAINERS + "U9/MobileMail.app/MobileMail"}]
        cond = targets_from(procs_mail, []) == []
        json.dump({"result": {"apps": [{"bundleIdentifier": f"com.apple.a{i}", "url": f"file:///Applications/A{i}.app/"} for i in range(3)]
                              + [{"bundleIdentifier": "com.VirZOOM.FlyVP", "url": "file:///A/FLY.app/"}]}}, open(g["APPS_CACHE"], "w"))
        cond = cond and targets_from(procs_mail, []) == []
        print(("PASS  " if cond else "FAIL  ") + "decoy: a cached list without Apple's apps never lets Reset close Mail"); fails += 0 if cond else 1
        _last["id"] = None
        g["running_targets"] = lambda roots, budget, dev=None: [(7, "file:///A/FLY.app/FLY")]
        ok, msg = reset_for_guest()
        cond = not ok and "FLY" in msg
        print(("PASS  " if cond else "FAIL  ") + "decoy: an app still running after Reset is named, never Ready"); fails += 0 if cond else 1
    finally:
        g.update(saved)
    return fails


def js_syntax_error(doc):
    """Compile every <script> in the page with JavaScriptCore (osascript -l JavaScript). Returns ''
    when all parse, the error text when one does not, or 'NOT MEASURED' when there is no runner."""
    import re as _re, tempfile
    if not os.path.exists("/usr/bin/osascript"):
        return "NOT MEASURED"
    for src in _re.findall(r"<script>(.*?)</script>", doc, _re.S):
        with tempfile.NamedTemporaryFile("w", suffix=".js", delete=False) as t:
            t.write(src)
        probe = ("ObjC.import('Foundation');var s=$.NSString.stringWithContentsOfFileEncodingError("
                 + json.dumps(t.name) + ",4,null).js;try{new Function(s);'OK'}catch(e){'ERR '+e}")
        r = subprocess.run(["/usr/bin/osascript", "-l", "JavaScript", "-e", probe], capture_output=True, text=True, timeout=20)
        os.remove(t.name)
        out = (r.stdout or r.stderr).strip()
        if out != "OK":
            return out or "no answer"
    return ""


def http_selftest():
    """Run the real handler on a spare port, in a throwaway config folder, with the headset
    calls replaced by stand-ins. Each rule has a case that must be refused and one that must pass."""
    import http.client, tempfile
    g = globals()
    saved = {n: g[n] for n in ("CFG", "APPS_CACHE", "LOG", "ICONS", "HERE", "lock_state", "launch", "reset_for_guest")}
    tmp = tempfile.mkdtemp(prefix="vp-demo-selftest-")
    json.dump({"result": {"apps": [{"bundleIdentifier": "com.VirZOOM.FlyVP", "name": "FLY",
               "url": "file:///private/var/containers/Bundle/Application/A/FLY.app/"}]}}, open(os.path.join(tmp, "apps.json"), "w"))
    g.update(CFG=tmp, APPS_CACHE=os.path.join(tmp, "apps.json"), LOG=os.path.join(tmp, "launches.log"),
             ICONS=os.path.join(tmp, "icons"), HERE=os.path.join(tmp, "bundle"), lock_state=lambda: "ready", launch=lambda b: (True, "opened"),
             reset_for_guest=lambda: (True, "reset"))
    os.makedirs(os.path.join(tmp, "bundle"))                     # apart from CFG, so the check cannot pass by accident
    open(os.path.join(tmp, "bundle", "app-icon-180.png"), "wb").write(b"\x89PNG-bundled")
    _pair.update(code=None, until=0.0); _tries.clear()
    srv = ThreadingHTTPServer(("127.0.0.1", 0), H)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    port, fails = srv.server_address[1], 0

    def req(method, path, headers=None, body=None):
        c = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
        c.request(method, path, body=body, headers=headers or {})
        r = c.getresponse(); data = r.read(); c.close()
        return r.status, dict(r.getheaders()), data

    def check(name, cond):
        nonlocal fails
        print(("PASS  " if cond else "FAIL  ") + name)
        fails += 0 if cond else 1

    try:
        st, h, _ = req("GET", "/")
        check("decoy: an unpaired visitor is sent to /pair", st == 302 and h.get("Location") == "/pair")
        st, _, _ = req("GET", "/pair", {"Host": "rebind.example:8787"})
        check("decoy: a page reached through another domain's name is refused (421)", st == 421)
        st, _, _ = req("POST", "/pair", {"Host": "rebind.example", "X-VPD": "1"}, json.dumps({"code": "000000"}))
        check("decoy: a pairing attempt through another domain's name is refused (421)", st == 421)
        st, _, _ = req("GET", "/pair", {"Host": f"localhost:{port}"})
        check("control: localhost with a port is this Mac's own name", st == 200)
        check("control: this Mac's bracketed IPv6 loopback and Bonjour name pass the Host check",
              host_ok("[::1]:8787") and any(host_ok(n + ":8787") for n in allowed_hosts() if n.endswith(".local")))
        check("decoy: an empty Host is refused", not host_ok(""))
        check("control: a bracketed IPv6 Host with a zone suffix is matched without the zone",
              host_ok("[::1%25lo0]:8787"))
        saved_h = dict(_hosts)
        _hosts.update(at=0.0)
        allowed_hosts(discover=lambda: ({"test-mac.local"}, set()))
        _hosts.update(at=0.0)
        allowed_hosts(discover=lambda: (set(), set()))          # the name lookup fails once
        check("control: a failed name lookup keeps the last good Bonjour name", host_ok("test-mac.local:8787"))
        _hosts.update(saved_h)
        check("decoy: with no engine on record, the key is never sent", not engine_is_ours(port))
        st, _, _ = req("GET", "/status")
        check("decoy: an unpaired status read is refused (401)", st == 401)
        st, _, _ = req("POST", "/launch?b=com.VirZOOM.FlyVP", {"X-VPD": "1"})
        check("decoy: an unpaired launch is refused (401)", st == 401)
        code, _ = pair_code()
        st, h, _ = req("POST", "/pair", {"X-VPD": "1", "Content-Type": "application/json"}, json.dumps({"code": code}))
        cookie = (h.get("Set-Cookie") or "").split(";")[0]
        check("control: the right code pairs and sets an HttpOnly cookie",
              st == 200 and cookie.startswith("vpd=") and "HttpOnly" in (h.get("Set-Cookie") or ""))
        st, _, _ = req("POST", "/pair", {"X-VPD": "1"}, json.dumps({"code": code}))
        check("decoy: a used code cannot pair a second device", st == 403)
        st, _, d = req("GET", "/status", {"Cookie": cookie})
        check("control: a paired device reads status", st == 200 and b"ready" in d)
        st, _, _ = req("POST", "/launch?b=com.VirZOOM.FlyVP", {"Cookie": cookie})
        check("decoy: a launch without X-VPD (a cross-site form) is refused (403)", st == 403)
        st, _, d = req("POST", "/launch?b=com.VirZOOM.FlyVP", {"Cookie": cookie, "X-VPD": "1"})
        check("control: a paired launch with X-VPD goes through", st == 200 and b'"ok": true' in d)
        st, _, _ = req("POST", "/launch?b=a;rm", {"Cookie": cookie, "X-VPD": "1"})
        check("decoy: a bad bundle id is refused (400)", st == 400)
        st, _, _ = req("GET", "/admin/code", {"Cookie": cookie})
        check("decoy: a paired phone cannot read the pairing code", st == 404)
        st, _, d = req("GET", "/admin/code", {"X-Key": key()})
        check("control: the Mac (X-Key) reads a fresh six-digit code", st == 200 and len(json.loads(d)["code"]) == 6)
        st, h, _ = req("GET", "/?k=" + key())
        check("decoy: a link carrying the master key is NOT accepted (sent to /pair)",
              st == 302 and h.get("Location") == "/pair" and not h.get("Set-Cookie"))
        _tries.clear()
        good = pair_code()[0]
        wrong = f"{(int(good) + 1) % 1000000:06d}"          # never equal to the live code
        for _ in range(4):
            req("POST", "/pair", {"X-VPD": "1"}, json.dumps({"code": wrong}))
        st, _, d = req("POST", "/pair", {"X-VPD": "1"}, json.dumps({"code": wrong}))
        check("decoy: the fifth wrong code burns the code for everyone", st == 403 and b"new one" in d)
        st, _, d = req("POST", "/pair", {"X-VPD": "1"}, json.dumps({"code": good}))
        check("decoy: the burned code no longer pairs", st == 403)
        st, _, d = req("POST", "/pair", {"X-VPD": "1"}, json.dumps({"code": pair_code()[0]}))
        check("decoy: after five wrong tries from one address even a fresh code waits", st == 403 and b"Too many" in d)
        _tries.clear()
        st, _, d = req("POST", "/admin/handoff", {"X-Key": key()})
        t = json.loads(d)["t"]
        st, h, _ = req("GET", "/pair?t=" + t)
        check("control: a one-time handoff link pairs and sets the cookie",
              st == 302 and (h.get("Set-Cookie") or "").startswith("vpd="))
        st, h, _ = req("GET", "/pair?t=" + t)
        check("decoy: the same handoff link cannot pair twice", st == 200 and not h.get("Set-Cookie"))
        st, _, _ = req("POST", "/launch?b=com.example.notinstalled", {"Cookie": cookie, "X-VPD": "1"})
        check("decoy: an app that is neither a tile nor installed is refused (403)", st == 403)
        st, _, _ = req("POST", "/launch?b=-x", {"Cookie": cookie, "X-VPD": "1"})
        check("decoy: an option-shaped id is refused (400)", st == 400)
        _tries.clear()
        code = pair_code(rotate=True)[0]
        st, h, _ = req("POST", "/pair", {"X-VPD": "1"}, json.dumps({"code": code[:3] + " " + code[3:]}))
        cookie2 = (h.get("Set-Cookie") or "").split(";")[0]
        check("control: the code typed with the space the menu shows ('123 456') pairs", st == 200 and cookie2.startswith("vpd="))
        st, h, _ = req("GET", "/pair", {"Cookie": cookie2})
        check("control: a paired device that opens /pair is sent to the remote", st == 302 and h.get("Location") == "/")
        st, h, _ = req("GET", "/pair")
        check("decoy: an unpaired /pair visit gets the code form", st == 200)
        dp = devices_path()
        before = (os.stat(dp).st_ino, os.stat(dp).st_mtime_ns)
        req("GET", "/status", {"Cookie": "vpd=not-a-real-token"})
        check("decoy: an unknown cookie never rewrites the paired list", (os.stat(dp).st_ino, os.stat(dp).st_mtime_ns) == before)
        req("GET", "/status", {"Cookie": cookie2})
        mid = (os.stat(dp).st_ino, os.stat(dp).st_mtime_ns)
        req("GET", "/status", {"Cookie": cookie2})
        check("control: two reads in the same minute rewrite the paired list at most once",
              (os.stat(dp).st_ino, os.stat(dp).st_mtime_ns) == mid)
        check("control: the paired list is created private (0600)", (os.stat(dp).st_mode & 0o777) == 0o600)
        local = next((n for n in allowed_hosts() if n.endswith(".local")), None)
        if local:
            st, _h, body = req("POST", "/admin/handoff?for=qr", {"X-Key": key()})
            t = json.loads(body)["t"]
            st, h, _ = req("GET", "/pair?t=" + t, {"Host": f"{local}:8787"})
            check("control: the QR link pairs through this Mac's Bonjour name (what the phone camera opens)",
                  st == 302 and (h.get("Set-Cookie") or "").startswith("vpd=") and json.loads(body)["seconds"] == 600)
            check("control: the QR answer carries the live six-digit code and its time left",
                  len(json.loads(body).get("code", "")) == 6 and json.loads(body).get("code_seconds", 0) > 0)
        else:
            print("NOT MEASURED  QR link through the Bonjour name: this Mac reported no .local name")
        st, _, d = req("GET", "/catalog", {"Cookie": cookie2})
        check("control: the catalog lists the demo sections", st == 200 and json.loads(d)["sections"])
        g["launch"] = saved["launch"]; g["_launch"], _sl = (lambda b, e: (True, "opened")), g["_launch"]
        _last["id"] = None
        req("POST", "/launch?b=title:elevated", {"Cookie": cookie2, "X-VPD": "1"})
        st, _, d = req("GET", "/status", {"Cookie": cookie2})
        check("control: a launch marks its tile Last opened for every remote (the title key, not the TV app)",
              json.loads(d).get("last") == "title:elevated")
        g["launch"], g["_launch"] = (lambda b: (True, "opened")), _sl
        st, _, d = req("POST", "/reset", {"Cookie": cookie2, "X-VPD": "1"})
        check("control: a paired Reset reaches the engine", st == 200 and b'"ok": true' in d)
        st, _, d = req("GET", "/app-icon-180.png")
        check("control: the Home Screen icon the app bundle carries is served (no Homebrew tool needed)",
              st == 200 and d == b"\x89PNG-bundled")
        st, h, _ = req("GET", "/manifest.webmanifest")
        check("control: the Home Screen manifest is served", st == 200 and "manifest" in (h.get("Content-Type") or ""))
    finally:
        srv.shutdown(); srv.server_close()
        g.update(saved); _pair.update(code=None, until=0.0); _tries.clear()
    return fails


AGENT = "com.jasonpaez.vp-demo"


def agent(port=8787):
    """Write the LaunchAgent and restart it. Interactive, not Background: a Background job is
    throttled when the Mac is loaded, and a demo tap must answer at once."""
    import plistlib
    path = os.path.expanduser(f"~/Library/LaunchAgents/{AGENT}.plist")
    me = os.path.expanduser("~/.local/bin/vp-demo")
    os.makedirs(CFG, exist_ok=True)
    with open(path, "wb") as f:
        plistlib.dump({"Label": AGENT, "ProgramArguments": ["/usr/bin/python3", me, "serve", "--port", str(port)],
                       "RunAtLoad": True, "KeepAlive": True, "ProcessType": "Interactive",
                       "StandardOutPath": os.path.join(CFG, "serve.log"),
                       "StandardErrorPath": os.path.join(CFG, "serve.log")}, f)
    dom = f"gui/{os.getuid()}"
    subprocess.run(["launchctl", "bootout", f"{dom}/{AGENT}"], capture_output=True)
    time.sleep(1)
    r = subprocess.run(["launchctl", "bootstrap", dom, path], capture_output=True, text=True)
    return r.returncode == 0, (r.stderr.strip() or path)


def main(argv):
    if not argv or argv[0] in ("-h", "--help"):
        print(__doc__); return 0
    cmd = argv[0]
    if cmd == "--selftest":
        return selftest()
    if cmd == "refresh":
        ok, m = refresh(); print(("OK " if ok else "FAIL ") + " ".join(m)); return 0 if ok else 1
    if cmd in ("code", "devices", "unpair-all", "reset"):
        import urllib.request
        port = 8787
        path = {"code": "/admin/code", "devices": "/admin/devices", "unpair-all": "/admin/unpair-all", "reset": "/reset"}[cmd]
        if not engine_is_ours(port):
            print(f"The process on port {port} is not this Mac's Vision Test Drive engine; the key was not sent."); return 1
        rq = urllib.request.Request(f"http://127.0.0.1:{port}{path}", method="POST" if cmd in ("unpair-all", "reset") else "GET",
                                    headers={"X-Key": key()})
        try:
            out = json.loads(urllib.request.urlopen(rq, timeout=RESET_BUDGET + 30).read())
        except Exception as e:
            print(f"The remote is not running on port {port}: {e}"); return 1
        if cmd == "code":
            print(f"Pairing code: {out['code']}  (good for {out['seconds_left'] // 60} more minutes, one device)")
        elif cmd == "devices":
            for d_ in out:
                print(f"  {d_['label']:24} paired {d_['paired']}  last used {d_['last']}")
            print(f"{len(out)} paired")
        else:
            print(out.get("message"))
            return 0 if out.get("ok", True) else 1
        return 0
    if cmd == "headsets":
        cur = os.environ.get("VP_DEMO_DEVICE") or (open(os.path.join(CFG, "device")).read().strip()
                                                  if os.path.exists(os.path.join(CFG, "device")) else "")
        hs = headsets()
        for i, n in hs:
            print(f"  {'*' if i == cur else ' '} {n:28} {i}")
        print(f"{len(hs)} headset{'s' if len(hs) != 1 else ''} paired with this Mac" + ("" if hs else " (pair one in Xcode, Devices and Simulators)"))
        return 0
    if cmd == "use":
        want = " ".join(argv[1:]).strip().lower()
        hit = [(i, n) for i, n in headsets() if want and (i.lower() == want or n.lower() == want)]
        if len(hit) != 1:
            print("Name one headset from `vp-demo headsets` (its name or identifier)."); return 1
        with device_turn() as mine:           # never switch headsets in the middle of a launch or reset
            if not mine:
                print("A launch or reset is running; try again in a moment."); return 1
            save_device(hit[0][0], hit[0][1])
            try:
                os.remove(APPS_CACHE)          # the app list belongs to the old headset
            except OSError:
                pass
        print(f"Using {hit[0][1]} ({hit[0][0]}); its app list is re-read when the remote next starts or on Refresh App List"); return 0
    if cmd == "agent":
        ok, m = agent(); print(("OK " if ok else "FAIL ") + m); return 0 if ok else 1
    if cmd == "list":
        apps = load_apps()
        for s, tiles in load_demo():
            print(f"[{s}]")
            for n, b in tiles:
                a = TITLES.get(b, (b, None))[0]
                print(f"  {n:24} {b}{'' if (not apps or a in apps) else '   (not installed)'}")
        print(f"{len(apps)} apps cached from the headset ({APPS_CACHE})")
        return 0
    if cmd == "launch":
        b = resolve(" ".join(argv[1:]), load_apps()) or " ".join(argv[1:])
        if not allowed(b):
            print(f"FAILED {b}: not a demo tile or an installed app"); return 1
        import urllib.request
        listening = subprocess.run(["/usr/sbin/lsof", "-nP", "-iTCP:8787", "-sTCP:LISTEN", "-t"],
                                   capture_output=True, text=True).stdout.strip()
        if not listening:
            ok, m = launch(b)                  # no engine running: launch directly
        elif not engine_is_ours():             # never hand the key to another program on the port
            print("FAILED: another program holds port 8787; the key was not sent"); return 1
        else:                                  # through the running engine, so its one-at-a-time rule holds
            try:
                rq = urllib.request.Request(f"http://127.0.0.1:8787/launch?b={urllib.parse.quote(b)}", method="POST",
                                            headers={"X-Key": key()})
                j = json.loads(urllib.request.urlopen(rq, timeout=120).read())
                ok, m = j.get("ok"), j.get("message", "")
            except Exception as e:             # the engine took the request: never launch a second time
                ok, m = False, f"no answer from the engine ({type(e).__name__}); the app may still be opening"
        print(("OPENED " if ok else "FAILED ") + b + ("" if ok else ": " + m)); return 0 if ok else 1
    if cmd == "serve":
        port = int(argv[argv.index("--port") + 1]) if "--port" in argv else 8787
        host = subprocess.run(["scutil", "--get", "LocalHostName"], capture_output=True, text=True).stdout.strip() or "localhost"
        os.makedirs(CFG, exist_ok=True)
        os.chmod(CFG, 0o700)                   # tokens, the key and the app list stay this user's
        for lf in (LOG, os.path.join(CFG, "serve.log")):
            try:
                if os.path.getsize(lf) > 1_000_000:
                    tail = open(lf, "rb").read()[-200_000:]
                    open(lf, "wb").write(tail[tail.find(b"\n") + 1:])
            except OSError:
                pass
        key()                                  # create or rotate the master key before the first request
        allowed_hosts()                        # read this Mac's names once, before the first request
        srv = ThreadingHTTPServer(("0.0.0.0", port), H)
        with open(os.path.join(CFG, "serve.pid"), "w") as f:
            f.write(str(os.getpid()))
        threading.Thread(target=warm_up, daemon=True).start()
        if "--parent-pid" in argv:
            parent = int(argv[argv.index("--parent-pid") + 1])

            def watch():
                # The Mac app is gone (crashed, force-quit, replaced by an update): stop too, so the
                # next copy of the app can bind the port instead of looping on "address in use".
                while True:
                    time.sleep(2)
                    try:
                        os.kill(parent, 0)
                    except ProcessLookupError:
                        os._exit(0)
                    except PermissionError:
                        pass
            threading.Thread(target=watch, daemon=True).start()
        print(f"Vision Pro demo remote: http://{host}.local:{port}/pair (pair with: vp-demo code)", flush=True)
        srv.serve_forever()
    print(__doc__); return 2


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))
